// -*-mode:c++; coding:utf-8-*-

#include <unistd.h>

#include <sstream>
#include <cstring>

#include "xbase_utility.hpp"
#include "logger.hpp"
#include "compressor.hpp"
#include "naming_strategy.hpp"		// for tablet_name

#include "protocol_struct.hpp"
#include "protocol_base.hpp"
#include "protocol.hpp"

namespace xbase
{
	namespace protocol
	{
		//
		// helper functions
		// 
		
		template<typename T>
		inline
		std::string to_string(const T &obj) {
			return obj.to_string();
		}

		template<>
		inline
		std::string to_string(const std::string &str) {
			return str;
		}

#define integer_to_string(type)				\
		template<>				\
		inline					\
		std::string to_string(const type &n) {	\
			std::ostringstream out;		\
			out << n;			\
			return out.str();		\
		}

		integer_to_string(uint32_t)

#undef integer_to_string
		
		template<typename elemT,
			 typename charT, typename traits>
		inline
		std::basic_ostream<charT, traits> &
		operator<<(std::basic_ostream<charT, traits> &out,
			   const std::vector<elemT> &vec)
		{
			typedef typename std::vector<elemT>::const_iterator cit_type;
			out << "[";
			const cit_type end = vec.end();
			cit_type it = vec.begin();
			if(it != end)
			{
				out << to_string(*it);
				for(++it; it != end; ++it)
				{
					out << ", " << to_string(*it);
				}
			}
			out << "]";
			return out;
		}

		//
		// protocol_base.hpp
		//

		static
		const char * const type_name_vector[COMMAND_COUNT] = {
			// 1. 查询根子表信息
			"ROOT_META_REQ",        //0x0000,
			"ROOT_META_RESP",       //0x0001,
			// 2. 向主服务器查询某个表的结构信息
			"TABLE_INFO_REQ",       //0x0002,
			"TABLE_INFO_RESP",      //0x0003,
			// 3. 创建表
			"CREATE_TABLE_REQ",     //0x0004,
			"CREATE_TABLE_RESP",    //0x0005,
			// 4. 为某个表添加列族
			"ADD_FAMILY_REQ",       //0x0006,
			"ADD_FAMILY_RESP",      //0x0007,
			// 5. 为某个表删除列族
			"DELETE_FAMILY_REQ",    //0x0008,
			"DELETE_FAMILY_RESP",   //0x0009,
			// 6. 添加某个表的组
			"ADD_GROUP_REQ",        //0x000A,
			"ADD_GROUP_RESP",       //0x000B,
			// 7. 删除某个表的组
			"DELETE_GROUP_REQ",     //0x000C,
			"DELETE_GROUP_RESP",    //0x000D,
			// 8. 删除表
			"DELETE_TABLE_REQ",     //0x000E,
			"DELETE_TABLE_RESP",    //0x000F,
			// 1. 读记录
			"READ_REQ",             //0x0010,
			"READ_RESP",            //0x0011,
			// 2. 写记录
			"WRITE_REQ",            //0x0012,
			"WRITE_RESP",           //0x0013,
			// 3. 扫描/遍历记录
			"SCAN_OPEN_REQ",        //0x0014,
			"SCAN_OPEN_RESP",       //0x0015,
			"SCAN_NEXT_REQ",        //0x0016,
			"SCAN_NEXT_RESP",       //0x0017,
			"SCAN_CLOSE_REQ",       //0x0018,
			"SCAN_CLOSE_RESP",      //0x0019,
			// 1. 启动时，向主服务器发送报到消息
			"TABLET_SERVER_REPORT", //0x001A,
			"BAD_TYPE_CODE",
			// 2. 运行过程中，向主服务器发送心跳信息
			"HEARTBEAT_REQ",        //0x001C,
			"BAD_TYPE_CODE",
			// 3. 主服务器向子表服务器发送加载子表命令
			"LOAD_TABLET_REQ",      //0x001E,
			"LOAD_TABLET_RESP",     //0x001F,
			// 4. 删除子表
			"REMOVE_TABLET_REQ",    //0x0020,
			"REMOVE_TABLET_RESP",   //0x0021,
			// 5. 子表分裂后，向主服务器报告哪个子表分裂成了哪两个子表
			"TABLET_SPLIT_REPORT",  //0x0022,
			"BAD_TYPE_CODE",
			// 6. 子表服务器退出时，向主服务器发送下线消息
			"TABLET_SERVER_QUIT",   //0x0024,
			"BAD_TYPE_CODE",
			//7. 主服务器要求子表服务器重启
			"TABLET_SERVER_RESTART",	//0x0026,
			"BAD_TYPE_CODE",
			// 列子表
			"LIST_TABLET_REQ",		// 0x0028
			"LIST_TABLET_RESP",		// 0x0029
			// 列表
			"LIST_TABLE_REQ",		// 0x002A
			"LIST_TABLE_RESP",		// 0x002B
			// 批量写
			"BATCH_WRITE_REQ",		// 0x002C
			"BATCH_WRITE_RESP",		// 0x002D
		};

		const char *type_name(type_code type) {
			const char * const bad_type_code = "BAD_TYPE_CODE";
			return (type < COMMAND_COUNT)
				? type_name_vector[type]
				: bad_type_code;
		}

		std::string packet::to_string() const {
			std::ostringstream out;
			out << "{magic:" << "0x" << std::hex << m_magic << std::dec
			    << "; version:" << (uint32_t)m_version
			    << "; type:" << "0x" << std::hex << m_type << std::dec
			    << "(" << type_name(static_cast<type_code>(m_type))
			    << "); sequence_number:" << m_sequence_number
			    << ((m_sequence_number == P_BAD_SEQUENCE) ? "(bad sequence)" : "")
			    << "; data_length:" << m_data_length
			    << "}";
			return out.str();
		}

		boost::system::error_code validate(const char *begin,
						   const char **end,
						   std::size_t *bytes_to_read,
						   std::size_t *id,
						   boost::system::error_code &ec)
		{
			const uint16_t magic_number_encoded
				= ::kb::to_byteorder_cast<byteorder>(P_MAGIC_NUMBER);
			const std::size_t header_size = packet::HEADER_SIZE;
			const std::size_t buffer_size = *end - begin;
			if(buffer_size < header_size)
			{
				*bytes_to_read = header_size - buffer_size;
				ec = error::header_incomplete;
			}
			else if(std::memcmp(&magic_number_encoded, begin, sizeof(magic_number_encoded)) != 0)
			{
				*bytes_to_read = 0;
				ec = error::bad_magic;
			}
			else
			{
				uint32_t sequence_number = 0;
				uint32_t data_length = 0;
				iarchive in(begin + packet::POS_SEQUENCE_NUMBER,
						 *end);
				const bool ok = (in.load(&sequence_number) &&
						 in.load(&data_length));
				assert(ok); (void)ok;

				const std::size_t packet_size = header_size + data_length;
				if(buffer_size < packet_size)
				{
					*bytes_to_read = packet_size - buffer_size;
					ec = error::packet_incomplete;
				}
				else
				{
					*end = begin + packet_size;
					*bytes_to_read = 0;
					*id = sequence_number;
					ec = boost::system::error_code();
				}
			}
			return ec;
		}

		// 
		// protocol_struct.hpp
		//
		
		std::string tablet_info::to_string() const {
			std::ostringstream out;
			out << "{tablet_id: " << m_tablet_id
			    << "; table_name: " << m_table_name
			    << "; start_row: " << dump_string_hex(m_start_row)
			    << "; end_row: " << dump_string_hex(m_end_row)
			    << "; status: 0x" << std::hex << m_status << std::dec
			    << "("
			    << (TabletStatusGetOnline(m_status) ? "online" : "offline") << " | "
			    << (TabletStatusGetSplitting(m_status) ? "splitted" : "non-splitted") << " | "
			    << (TabletStatusGetDeleted(m_status) ? "deleted" : "non-deleted")
			    << ")}";
			return out.str();
		}

		std::string tablet_info::tablet_name() const {
			return naming_strategy::tablet_name(m_tablet_id,
							    m_table_name,
							    m_start_row,
							    m_end_row);
		}

		bool tablet_info::is_root_meta_tablet() const {
			return (m_tablet_id == 0) &&
				(m_table_name == ROOT_META_TABLET_NAME);
		}

		std::string server_address::to_string() const {
			std::ostringstream out;
			out << "{hostname: " << m_hostname
			    << "; port: " << m_port
			    << "; startcode: " << m_startcode
			    << "}";
			return out.str();
		}

		std::string key::to_string () const {
			std::ostringstream out;
			out << "{row_name:" << dump_string_char(m_row_name)
			    << "; column_name:" << dump_string_char(m_column_name)
			    << "; timestamp:" << m_timestamp
			     << '(' << dump_timestamp(m_timestamp)
			    << ")}";
			return out.str();
		}

		std::string record::to_string () const {
			std::ostringstream out;
			out << "{key:" << m_key.to_string()
			    << "; value:" << dump_string_hex(m_value)
			    << "}";
			return out.str();
		}

		std::string row_mutation_data::to_string() const {
			std::ostringstream out;
			out << "{row: " << dump_string_char(m_row_name)
			    << "; groups: ";
			
			typedef row_mutation_data::groups::const_iterator g_iter;
			typedef row_mutation_data::columns::const_iterator c_iter;
				
			const g_iter end_group = m_groups.end();
			for(g_iter iter_group = m_groups.begin();
			    iter_group != end_group;
			    ++ iter_group)
			{
				out << "{group_name: " << iter_group->first
				    << "; columns: [";
				const c_iter end_column = iter_group->second.end();
				for(c_iter iter_column = iter_group->second.begin();
				    iter_column != end_column;
				    ++ iter_column)
				{
					out << "{column_name: " << iter_column->m_column_name
					    << "; timetamp: " << iter_column->m_timestamp
					    << "; value: " << iter_column->m_value
					    << "}";
				}
				out << "]}";
			}
			out << "}";
			return out.str();
		}
		
		void row_mutation_data::fill_record_vector(record_vector &vec,
							   uint64_t timestamp) const {
			typedef groups::const_iterator g_iter;
			typedef columns::const_iterator c_iter;
				
			const g_iter end_group = m_groups.end();
			for(g_iter iter_group = m_groups.begin();
			    iter_group != end_group;
			    ++ iter_group)
			{
				const c_iter end_column = iter_group->second.end();
				for(c_iter iter_column = iter_group->second.begin();
				    iter_column != end_column;
				    ++ iter_column)
				{
					vec.push_back(
						make_record(m_row_name,
							    iter_column->m_column_name,
							    ((iter_column->m_timestamp == NO_TIMESTAMP)
							     ? timestamp : iter_column->m_timestamp),
							    iter_column->m_value)
						);
				}
			}
		}

		std::string row_mutation_data_for_tabletserver::to_string() const {
			const std::size_t max_dump_size = 1024;
			std::ostringstream out;
			out << "{row;";
			for(group_vector::const_iterator g_it = m_groups.begin();
			    g_it != m_groups.end(); ++ g_it)
			{
				out << " {group_name: " << g_it->first
				    << "; group_data bytes: " << boost::asio::buffer_size(g_it->second)
				    << "; group_data: " << dump_string_hex(boost::asio::buffer_cast<const char*>(g_it->second),
									   (std::min)(boost::asio::buffer_size(g_it->second), max_dump_size))
				    << "}";
			}
			out << "}";
			return out.str();
		}

		bool group_options::valid() const {
			return compressor::valid(static_cast<compressor::compressor_code>(m_compress_method)) &&
				(m_cache_rate <= 100) &&
				(m_block_size != 0) &&
				(m_max_num_sstable > 1) &&
				(m_max_size_sstable != 0) &&
				(m_max_size_memtable != 0);
		}

		std::string group_options::to_string () const {
			std::ostringstream out;
			out << "{compress_method: "
			    << "0x" << std::hex << m_compress_method << std::dec
			    << "(" << compressor::compressor_name(static_cast<compressor::compressor_code>(m_compress_method))
			    << "); cache_rate: " << m_cache_rate << "%"
			    << "; block_size: " << m_block_size
			    << "; max_num_sstable: " << m_max_num_sstable
			    << "; max_size_sstable: " << m_max_size_sstable
			    << "; max_size_memtable: " << m_max_size_memtable
			    << "}";
			return out.str();
		}

		std::string group_info::to_string () const {
			std::ostringstream out;
			out << "{group_name: " << m_group_name
			    << "; options: " << m_options.to_string()
			    << "}";
			return out.str();
		}

		std::string family_options::to_string () const {
			std::ostringstream out;
			out << "{max_num_version: " << m_max_num_version
			    << (m_max_num_version == 0 ? "(unlimited)" : "")
			    << "; max_ttl: " << m_max_ttl
			    << (m_max_ttl == 0 ? "(unlimited)" : "")
			    << "; dead_time: " << m_dead_time
			    << (m_dead_time == 0 ? "(unlimited)" : "")
			    << "}";
			return out.str();
		}

		std::string family_info::to_string () const {
			std::ostringstream out;
			out << "{family_name: " << m_family_name
			    << "; group_name: " << m_group_name
			    << "; options: " << m_options.to_string()
			    << "}";
			return out.str();
		}

		// TODO: 检查有待完善
		bool table_info::valid() const {
			// 表名不能为空，且不能含有/、空格、,、及其它目录名中不允许出现的字符
			if(m_table_name.empty() ||
			   (m_table_name.find_first_of("/ ,'\"") != std::string::npos))
				return false;

			if(m_groups.empty() || m_families.empty())
				return false;

			//遍历每个列族，查看它的组是否存在
			const family_info_vector::const_iterator it_end = m_families.end();
			for(family_info_vector::const_iterator it = m_families.begin();
			    it != it_end; ++ it)
			{
				if(get_group_info(it->m_group_name) == NULL)
					return false;
			}
			return true;
		}

		std::string table_info::to_string () const {
			std::ostringstream out;
			out << "{table_name: " << m_table_name
			    << "; groups: " << m_groups
			    << "; families: " << m_families
			    << "}";
			return out.str();
		}

		std::string row_data::to_string() const {
			std::ostringstream out;
			out << m_records;
			return out.str();
		}

		std::string row_data_v2::to_string() const {
			std::ostringstream out;
			out << m_mutation_data->to_string();
			return out.str();
		}

		void server_load::update_meminfo() {
			int fd = ::open("/proc/meminfo", O_RDONLY);
			if(fd < 0)
			{
				LOG(ERROR, "open \"/proc/meminfo\" failed, errno = "
				    << errno);
				return;
			}
			const int size = 1024;
			char buffer[size] = {0};
			ssize_t ret = ::read(fd, buffer, size);
			::close(fd);
			if(ret < 0)
			{
				LOG(ERROR, "read '/proc/meminfo' failed, errno = "
				    << errno);
				return;
			}
				
			const int char_per_line = 25;
			const int bytes_per_line = char_per_line + 1;
			const char *pos_total = buffer + 12;
			const char *pos_free = buffer + bytes_per_line + 12;
			const char *pos_buffers = buffer + 2 * bytes_per_line + 12;
			const char *pos_cached = buffer + 3 * bytes_per_line + 12;
			std::sscanf(pos_total, "%d", &m_total_memory);
			m_free_memory = 0;
			uint32_t free = 0;
			std::sscanf(pos_free, "%d", &free);
			m_free_memory += free;
			std::sscanf(pos_buffers, "%d", &free);
			m_free_memory += free;
			std::sscanf(pos_cached, "%d", &free);
			m_free_memory += free;
				
			m_total_memory = m_total_memory << 10;
			m_free_memory = m_free_memory << 10; // * 1024
		}

		void server_load::update_cpuinfo() {
			// int fd = ::open("/proc/loadavg", O_RDONLY);
			// //if(fd == -1)
			// const int size = 64;
			// char buffer[size];
			// ::read(fd, buffer, size);
			// ::close(fd);

			// double cpus_load;
			// std::sscanf(buffer, "%f", &cpus_load);
			// m_cpus_load = (uint32_t)(100.0 * cpus_load);
			std::ifstream in("/proc/loadavg");
			double cpus_load;
			in >> cpus_load;
			in.close();
			m_cpus_load = static_cast<uint32_t>(cpus_load * 100.0);
		}

		void server_load::update_loadinfo() {
			// 1. memory load
			double memory_used = static_cast<double>(m_memory_table_size + m_cached_block_size);
			double memory_load = ((memory_used / static_cast<double>(m_total_memory))
					      + (static_cast<double>(K) / (m_total_memory - memory_used))) * K3;

			// 2. calc prev query load
			m_prev_query_load = m_query_load * A / 100;

			int done_message = (m_input_message_number - m_left_input_message_number);

			// 3. update capability
			m_capability = (std::max<int>)((int)m_capability, done_message);

			// 4. query load with new prev load and capability
			double query_load = (static_cast<double>(m_prev_query_load) / 10000.0)
				+ ((static_cast<double>(m_input_message_number) / static_cast<double>(m_capability))
				   + (static_cast<double>(m_left_input_message_number) / static_cast<double>(K2)));

			// 5. total load
			double load = memory_load + query_load;

			// save values
			m_memory_load = static_cast<uint32_t>(10000 * memory_load);
			//m_prev_query_load = static_cast<uint32_t>(10000 * prev_query_load);
			m_query_load = static_cast<uint32_t>(10000 * query_load);
			m_load = static_cast<uint32_t>(10000 * load);
		}

		std::string server_load::to_string() const {
			std::ostringstream out;
			out << "{total_memory:" << m_total_memory
			    << "; free_memory:" << m_free_memory
			    << "; cpus_load:" << m_cpus_load
			    << "; input_message_number:" << m_input_message_number
			    << "; left_input_message_number:" << m_left_input_message_number
			    << "; output_message_number:" << m_output_message_number
			    << "; left_output_message_number:" << m_left_output_message_number
			    << "; tablet_number:" << m_tablet_number
			    << "; memory_table_size:" << m_memory_table_size
			    << "; cached_block_size:" << m_cached_block_size
			    << "; cached_block_number:" << m_cached_block_number
			    << "; capability:" << m_capability
			    << "; prev_query_load:" << m_prev_query_load
			    << "; query_load:" << m_query_load
			    << "; memory_load:" << m_memory_load
			    << "; load:" << m_load
			    << "}";
			return out.str();
		}

		std::string tablet_server_info::to_string() const {
			std::ostringstream out;
			out << "{address: " << m_address.to_string()
			    << "; load: " << m_load.to_string()
			    << "}";
			return out.str();
		}

		//
		// protocol_packet.hpp
		//

		std::string request::to_string() const {
			std::ostringstream out;
			out << "{program_version: " << m_program_version
			    << "}";
			return out.str();
		}
		
		std::string response::to_string() const {
			std::ostringstream out;
			out << "{error_code: " << error_code()
			    << "}";
			return out.str();
		}
		
		std::string root_meta_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
			    << "; tablet: " << m_tablet.to_string()
			    << "; server_address: " << m_server_address.to_string()
			    << "}";
			return out.str();
		}

		std::string write_request::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; tablet_name:" << m_tablet_name
			    << "; records:" << m_records
			    << "}";
			return out.str();
		}

		std::string write_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
			    << "; tablet_name:" << m_tablet_name
			    << "; record_number:" << m_record_number
			    << "}";
			return out.str();
		}

		std::string write_request_v2::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; tablet_name:" << m_tablet_name
			    << "; records:" << m_mutation_data->to_string()
			    << "}";
			return out.str();
		}

		std::string write_request_v2_for_tabletserver::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; tablet_name:" << m_tablet_name
			    << "; records:" << m_mutation_data.to_string()
			    << "}";
			return out.str();
		}

		std::string table_info_request::to_string () const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; table_name: " << m_table_name
			    << "}";
			return out.str();
		}

		std::string table_info_response::to_string () const {
			std::ostringstream out;
			out << "{" << response::to_string()
			    << "; table_info: " << m_table_info.to_string()
			    << "}";
			return out.str();
		}

		std::string scan_open_request::to_string () const {
			std::ostringstream out;
			out << "{" << request::to_string()
				<< "; tablet_name: " << m_tablet_name
				<< "; start_row_name: " << m_start_row
				<< "; end_row_name: " << m_end_row
				<< "; columns: " << m_column_names
				<< "; lastest_time_stamp: " << m_latest_time_stamp
				<< "; oldest_time_stamp: " << m_oldest_time_stamp
				<< "; ttl: " << m_ttl
				<< "; version_number: " << m_version_number
				<< "; options: " << m_options
				<< "; num_row: " << m_num_rows
				<< "; row_offset: " << m_row_offset
				<< "; row_count: " << m_row_count
				<< "}";
			return out.str();
		}

		std::string scan_open_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
				<< "; tablet_name: " << m_tablet_name
				<< "; scanner_id: " << m_scanner_id
				<< "; row_offset: " << m_row_offset
				<< "}";
			return out.str();
		}

		std::string scan_next_request::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
				<< "; tablet_name: " << m_tablet_name
				<< "; scanner_id: " << m_scanner_id
				<< "}";
			return out.str();
		}

		std::string scan_next_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
				<< "; scanner_id: " << m_scanner_id
				<< "; tablet_name: " << m_tablet_name
				<< "; record_number: " << m_records.size()
				<< "}";
			return out.str();
		}

		std::string scan_close_request::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
				<< "; tablet_name: " << m_tablet_name
				<< "; scanner_id: " << m_scanner_id
				<< "}";
			return out.str();
		}

		std::string scan_close_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
				<< "; tablet_name: " << m_tablet_name
				<< "; scanner_id: " << m_scanner_id
				<< "; row_count: " << m_row_count
				<< "}";
			return out.str();
		}

		std::string batch_write_request::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; tablet_name: " << m_tablet_name
			    << "; rows: " << m_rows
			    << "}";
			return out.str();
		}

		std::string batch_write_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
			    << "; tablet_name: " << m_tablet_name
			    << "; status: " << m_status
			    << "}";
			return out.str();
		}

		std::string batch_write_request_v2::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; tablet_name: " << m_tablet_name
			    << "; rows: " << m_rows
			    << "}";
			return out.str();
		}

		std::string batch_write_response_v2::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
			    << "; success_row: " << m_success_row
			    << "}";
			return out.str();
		}

		std::string tablet_server_report::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; server_address: " << m_server_address.to_string()
			    << "}";
			return out.str();
		}

		std::string heartbeat_request::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; tablet_server_info: "
			    << m_tablet_server_info.to_string()
			    << "}";
			return out.str();
		}

		std::string load_tablet_request::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; tablet_info: " << m_tablet_info.to_string()
			    << "; table_info: " << m_table_info.to_string()
			    << "}";
			return out.str();
		}

		std::string load_tablet_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
			    << "; tablet_name: " << m_tablet_name
			    << "; server_address: " << m_server_address.to_string()
			    << "}";
			return out.str();
		}

		std::string remove_tablet_request::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; tablet_name: " << m_tablet_name
			    << "}";
			return out.str();
		}
		
		std::string remove_tablet_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
			    << "; tablet_name: " << m_tablet_name
			    << "}";
			return out.str();
		}

		std::string tablet_split_report::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; parent_tablet: " << m_parent_tablet.to_string()
			    << "; top_tablet: " << m_top_tablet.to_string()
			    << "; bottom_tablet: " << m_bottom_tablet.to_string()
			    << "}";
			return out.str();
		}

		std::string tablet_server_quit::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; server_address: " << m_server_address.to_string()
			    << "}";
			return out.str();
		}

		std::string list_tablet_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
			    << "; tablet_infos: " << m_tablet_infos
			    << "}";
			return out.str();
		}

		std::string create_table_request::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; table_info: " << m_table_info.to_string()
			    << "}";
			return out.str();
		}

		std::string create_table_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
			    << "; table_name: " << m_table_name
			    << "}";
			return out.str();
		}

		std::string delete_table_request::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; table_name: " << m_table_name
			    << "}";
			return out.str();
		}

		std::string delete_table_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
			    << "; table_name: " << m_table_name
			    << "}";
			return out.str();
		}

		std::string read_request::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; tablet_name: " << m_tablet_name
			    << "; key: " << m_key.to_string()
			    << "}";
			return out.str();
		}

		std::string read_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
			    << "; tablet_name: " << m_tablet_name
			    << "; records: " << m_records
			    << "}";
			return out.str();
		}

		std::string list_table_request::to_string() const {
			std::ostringstream out;
			out << "{" << request::to_string()
			    << "; start_table_name: " << m_start_table_name
			    << "; table_number: " << m_table_number
			    << "}";
			return out.str();
		}

		std::string list_table_response::to_string() const {
			std::ostringstream out;
			out << "{" << response::to_string()
			    << "; table_names: " << m_table_names
			    << "}";
			return out.str();
		}
		
	} // namespace protocol

} // namespace xbase

